package com.liusiqi;

import com.liusiqi.service.KafkaSender;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import javax.annotation.Resource;
import java.util.concurrent.CountDownLatch;

@SpringBootTest(classes = ClientApplication.class)
@RunWith(SpringJUnit4ClassRunner.class)
public class TaskKafka {

    //并发量
    private static final int USER_NUM = 1000;

    //倒计时器，用于模拟高并发
    private static CountDownLatch cdl = new CountDownLatch(USER_NUM);

    @Resource
    private KafkaSender kafkaSender;

    @Test
    public void testKafkaMq() throws Exception {
        //实例化USER_NUM个并发请求
        for (int i = 0; i < USER_NUM; i++) {
            new Thread(new UserRequest(i)).start();
            cdl.countDown();//倒计时减1
        }
        Thread.currentThread().sleep(30000);
    }

    /**
     * 内部类，用于模拟买票请求
     */
    public class UserRequest implements Runnable {

        private int id;

        public UserRequest(int id) {
            this.id = id;
        }
        @Override
        public void run() {
            try {
                //线程等待，等所以线程实例化完成
                cdl.await();
            } catch (Exception e) {
                e.printStackTrace();
            }
            String topic = "traffic-shopping-result";//主题
            String key = "user-"+id;//键
            String value = "123456="+System.currentTimeMillis();
            kafkaSender.messageSender(topic, key, value);
        }
    }
}
